package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"time"
)

// KeyValue
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

// ByKey for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

// Worker
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.
	for {
		reply := AskTask()
		if reply.AllFinished {
			return
		}
		if reply.AllGiven {
			time.Sleep(100 * time.Millisecond)
			continue
		}
		if reply.IsMap {
			encList := make(map[string]*json.Encoder)
			fileList := make(map[string]*os.File)
			for i := 0; i < reply.NReduce; i++ {
				fileName := fmt.Sprintf("mr-%v-%v", reply.TaskNumber, i)
				intFile, err := os.OpenFile(fileName, os.O_RDWR|os.O_CREATE, 0766)
				if err != nil {
					log.Fatalf("cannot open %v", fileName)
				}
				fileList[fileName] = intFile
				encList[fileName] = json.NewEncoder(fileList[fileName])
			}
			file, err := os.Open(reply.InputFileName)
			if err != nil {
				log.Fatalf("cannot open %v", reply.InputFileName)
			}
			content, err := ioutil.ReadAll(file)
			if err != nil {
				log.Fatalf("cannot read %v", reply.InputFileName)
			}
			file.Close()
			kva := mapf(reply.InputFileName, string(content))

			sort.Sort(ByKey(kva))
			for i := 0; i < len(kva); {
				j := i + 1
				for j < len(kva) && kva[j].Key == kva[i].Key {
					j++
				}
				var values []KeyValue
				for k := i; k < j; k++ {
					values = append(values, kva[k])
				}

				// here store intermediate k/v pairs to file
				fileName := fmt.Sprintf("mr-%v-%v", reply.TaskNumber, ihash(kva[i].Key)%reply.NReduce)
				for _, kv := range values {
					encList[fileName].Encode(&kv)
				}
				i = j
			}
			for i := 0; i < reply.NReduce; i++ {
				fileName := fmt.Sprintf("mr-%v-%v", reply.TaskNumber, i)
				fileList[fileName].Close()
			}
			FinishReport(true, reply.TaskNumber)
		} else {
			var intermediate []KeyValue
			outputName := fmt.Sprintf("mr-out-%v", reply.TaskNumber-1)
			outputFile, _ := ioutil.TempFile("", outputName)
			// outputFile, _ := os.OpenFile(outputName, os.O_RDWR|os.O_CREATE, 0766)
			decList := make(map[string]*json.Decoder)
			fileList := make(map[string]*os.File)
			for i := 0; i < reply.NMap; i++ {
				fileName := fmt.Sprintf("mr-%v-%v", i, reply.TaskNumber-1)
				intFile, err := os.Open(fileName)
				if err != nil {
					log.Fatalf("cannot open %v", fileName)
				}
				fileList[fileName] = intFile
				decList[fileName] = json.NewDecoder(fileList[fileName])
			}
			for i := 0; i < reply.NMap; i++ {
				fileName := fmt.Sprintf("mr-%v-%v", i, reply.TaskNumber-1)
				for {
					var kv KeyValue
					if err := decList[fileName].Decode(&kv); err != nil {
						break
					}
					intermediate = append(intermediate, kv)
				}
			}
			sort.Sort(ByKey(intermediate))

			for i := 0; i < len(intermediate); {
				j := i + 1
				for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
					j++
				}
				var values []string
				for k := i; k < j; k++ {
					values = append(values, intermediate[k].Value)
				}
				// ch := make(chan string)
				if _, err := fmt.Fprintf(outputFile, "%v %v\n", intermediate[i].Key, reducef(intermediate[i].Key, values)); err != nil {
					panic(err)
				}
				// output := reducef(intermediate[i].Key, values)
				// fmt.Fprintf(outputFile, "%v %v\n", intermediate[i].Key, output)

				// fmt.Printf("%v %v at task %v\n", intermediate[i].Key, output, reply.TaskNumber)
				// output := reducef(intermediate[i].Key, values)
				// fmt.Fprintf(outputFile, "%v %v\n", intermediate[i].Key, output)
				i = j
			}
			for i := 0; i < reply.NMap; i++ {
				fileName := fmt.Sprintf("mr-%v-%v", i, reply.TaskNumber-1)
				fileList[fileName].Close()
			}
			outputFile.Close()

			// fmt.Printf("I finished!!!!!! %v\n", reply.TaskNumber)
			os.Rename(outputFile.Name(), outputName)
			FinishReport(false, reply.TaskNumber)
			for i := 0; i < reply.NMap; i++ {
				fileName := fmt.Sprintf("mr-%v-%v", i, reply.TaskNumber-1)
				os.Remove(fileName)
			}
		}
		time.Sleep(100 * time.Millisecond)
	}

	// uncomment to send the Example RPC to the master.
	// CallExample()

}

// func doReduce(reducef func(string, []string) string, key string, values []string, ch chan string) error {
// 	output := reducef(key, values)
// 	ch <- output
// 	fmt.Println(output)
// 	return nil
// }

func FinishReport(isMap bool, taskNumber int) {
	args := FinishReportArgs{IsFinishMap: isMap, TaskNumber: taskNumber}
	reply := FinishReportReply{}
	call("Master.HandleFinishReport", &args, &reply)
}

func AskTask() *AskTaskReply {
	args := AskTaskArgs{}
	reply := AskTaskReply{}
	call("Master.GiveTask", &args, &reply)
	return &reply
}

// CallExample
// example function to show how to make an RPC call to the master.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Master.Example", &args, &reply)

	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)
}

//
// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
